/**  
  -- The Performance Driven Lime Project -- 

  A self adaptive tuple space for mobile agents and hosts by dynamic
  adapting tuple's replication policy.
  
  Copyright (C) 2008
  Xing Jiankuan, Pan Wenhui
  SAGroup of Tsinghua University

  This library is free software; you can redistribute it and/or
  modify it under the terms of the GNU Lesser General Public
  License as published by the Free Software Foundation; either
  version 2.1 of the License, or (at your option) any later version.

  This library is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received this copy of the GNU Lesser General Public
  License along with this library; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
 */
package sagroup.dracots;


import java.util.HashMap;

import java.util.HashSet;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.Set;
import java.util.TimerTask;
import java.util.logging.Logger;

import lime2.CommunicationMgr;
import lime2.Location;
import lime2.RemTupleTypeInfoOp;
import lime2.TupleID;
import sagroup.dracots.policy.PolicyRegistrar;
import sagroup.dracots.policy.RPTuple;
import sagroup.dracots.policy.ReplicationPolicy;
import sagroup.dracots.TupleType;

/**
 * @author Xing Jiankuan
 * 
 */
public class PolicyDecisionMaker extends TimerTask {

	private Period currentPeriod;
	private CostManager costQueue;
	private VersionManager versionManager;
	private RTUPManager rtupManager;
	
	private PolicyRegistrar policyRegistrar;
	private String TSName;
	private DracoTS dts;
	
	private Logger LOG = Logger.getLogger(this.getClass().getName());
	
	private CommunicationMgr cMgr = CommunicationMgr.get();
	
	

	private static boolean swtichOn = false;
	public static void setSwtichOn(boolean on)
	{
		swtichOn = on;
	}
	public PolicyDecisionMaker(DracoTS ts) {
		
		currentPeriod = Period.createPeriod(ts.getName());
		costQueue = new CostManager(DracoConstants.ADAPTIVE_CACULATION_RANGE,
				DracoConstants.RTUP_CACULATION_RANGE);
		costQueue.initialize(currentPeriod);
		rtupManager = new RTUPManager(DracoConstants.ADAPTIVE_CACULATION_RANGE,
				DracoConstants.RTUP_CACULATION_RANGE,ts.getLocalReplicaRegistrar(), ts.getPolicyRegistrar());
		versionManager = new VersionManager(costQueue);
		
		policyRegistrar = ts.getPolicyRegistrar();
		dts = ts;
		TSName = ts.getName();
	}
	
	private static HashMap<DracoTS,PolicyDecisionMaker>
	maps = new HashMap<DracoTS,PolicyDecisionMaker>();
	public static PolicyDecisionMaker getPolicyDecisionMaker(DracoTS ts)
	{
		PolicyDecisionMaker maker = null;
		if(!maps.containsKey(ts))
		{
		   maker =  new PolicyDecisionMaker(ts);
		   maps.put(ts, maker);
		}
		 else
			maker = maps.get(ts);
		
		return maker;
	}

	public synchronized void insertReadRecord(ShareTuple st) {
		costQueue.insertReadRecord(st, getCurrentPeriod());
	}


	public synchronized void insertUpdateRecord(TupleType type, long times) {
		Period p = getCurrentPeriod();
		costQueue.insertUpdateRecord(type, p, times);
		
	}



	public Period getCurrentPeriod() {
		return this.currentPeriod;
	}

	private synchronized void publishNewPeriod() {
		currentPeriod = Period.createPeriod(TSName);
		costQueue.initialize(currentPeriod);
		rtupManager.initialize(currentPeriod);
	}

	
	public void run() {

		if(currentPeriod == null)
			return;
		
		gatherAllRemoteInfo();
		
		resetStatics();
		
		
	    LOG.info("Policy Switch Info: " + costQueue.getCostString(currentPeriod));
		
		Set<TupleType> allTypes = new HashSet<TupleType>();
		allTypes.addAll(costQueue.getTypeSet());
		allTypes.addAll(rtupManager.getTypeSet());
		// log last period record
		for (TupleType type : allTypes)// all the types of DracoTS
		{
			if(type.equals(SystemTupleType.REPLICATION_POLICY))
				continue;
			
			double p = costQueue.getPValue(type,currentPeriod);
			double rtup = rtupManager.getRtupValue(type,currentPeriod);
			double MRtup = rtupManager.calMRtupValue(type,currentPeriod);
			double Mp = costQueue.calculateMp(type, currentPeriod);
			
			
			SwitchType st = getSwitchType(type);
			
			double mse1 = costQueue.getMSE(type); 
			double mse2 = rtupManager.getMSE(type);
			LOG.info("Original Policy Info: Period: "+ currentPeriod +" Type :" + type + " p : " + p
					+ " rtup : " + rtup);
					
			LOG.info("Policy Switch Info: Period: "+ currentPeriod +" Type :"+ type + " Mp : " + Mp
					+ " MP MSE: " + mse1 + " MRtup : " + MRtup + " MRtup MSE : " + mse2
					+" Policy Switch : " + st);
			
            if(swtichOn)
			switchPolicy(type, st);
		
		}

		publishNewPeriod();
		

	}

	private void resetStatics() {
		
		// NOTE: The remote type info may contain empty element when remote host inp the tuple first!!!
		RemoteTypeInfo[] infos = RemoteTypeInfo.getTypeRecordList(TSName);
		for(int i=0; i< infos.length;i++)
		{
			RemoteTypeInfo in = infos[i];
			TupleType t = in.getType();
			
		    //Set Update Count at this period ,and estimated tuples number
			//long deltaUpdateCount =  in.getTotalDelUpdateCount(); 
			long totalUpdateCount = in.getTotalUpdate();
			long estimatedTuples = in.getTotalEstimatedNum();
			costQueue.setUpdateInfo(t, currentPeriod,
					totalUpdateCount, estimatedTuples);
			
			//Set Unused Tuples Count;
			if(policyIsNotSO(t))
			{
				
				rtupManager.SetUnusedTuples(t,currentPeriod,estimatedTuples,
						dts.lrr.getAddedCountByType(t),
						dts.lrr.getDeletedCountByType(t));

			}
		}
	}
	private boolean policyIsNotSO(TupleType t) {
		ReplicationPolicy r = policyRegistrar.getReplicaPolicy(t, Location.HERE);
		if(r == null || r.equals(ReplicationPolicy.SO)) 
			return false;
		else
			return true;
	}

	private void switchPolicy(TupleType type, SwitchType st) {
		if (st == SwitchType.NONE)
			return;

		ReplicationPolicy rp = ReplicationPolicy.SO;
		
		if (st == SwitchType.FR_TO_SO || st == SwitchType.RC_TO_SO)
			rp = ReplicationPolicy.SO;
		else
		if (st == SwitchType.SO_TO_FR || st == SwitchType.RC_TO_FR)
			rp = ReplicationPolicy.FR;
		else
		if (st == SwitchType.SO_TO_RC || st == SwitchType.FR_TO_RC)
			rp = ReplicationPolicy.RC;
		RPTuple rt = new RPTuple(type,rp);
		dts.out(rt);
		LOG.info("����ת��: Type " + type + "has executed " + st + " .");
	}



	private static enum SwitchType {
		FR_TO_SO, FR_TO_RC, SO_TO_FR, SO_TO_RC, RC_TO_FR, RC_TO_SO, NONE
	}


	

	private SwitchType getSwitchType(TupleType type) {
		ReplicationPolicy rp = policyRegistrar.getReplicaPolicy(type,
				Location.HERE);
		if(rp == null)
		    rp = ReplicationPolicy.SO;
		if (rp.equals(ReplicationPolicy.FR)) // current policy FR
		{
			if (costQueue.needToSO(type))
				return SwitchType.FR_TO_SO;
			if (rtupManager.needToRC(type))
				return SwitchType.FR_TO_RC;
		}
		if (rp.equals(ReplicationPolicy.RC)) // current policy RC
		{
			if (costQueue.needToSO(type))
				return SwitchType.RC_TO_SO;
			if (rtupManager.needToFR(type))
				return SwitchType.RC_TO_FR;
		}

		if (rp.equals(ReplicationPolicy.SO)) // current policy SO
		{
			if (costQueue.needToREP(type)) {
				if (rtupManager.needToFR(type))
					return SwitchType.SO_TO_FR;
				else
					return SwitchType.SO_TO_RC;

			}
		}
		return SwitchType.NONE;
	}

	/**
	 * for test only
	 * 
	 * @return
	 * @author Xing Jiankuan
	 */
	Hashtable<TupleType, HashMap<Period, PeriodCost>> getCostsTable() {
		return costQueue.getCostsTable();
	}

	/**
	 * for test only
	 * 
	 * @return
	 * @author Xing Jiankuan
	 */
	Hashtable<TupleID, VersionRecord> getVersionRecords() {
		return versionManager.getVersionRecords();
	}

	/**
	 * for test only
	 * 
	 * @return
	 * @author Xing Jiankuan
	 */
	CostManager getCostQueue() {
		return this.costQueue;
	}

	

	public void handleTouch(ShareTuple t) {
		rtupManager.handleTouch(t, getCurrentPeriod());
	}

	
	
	/**
	 * Acquire the write and update information from a remote LimeServer
	 * @param target
	 * 			remote LimeServer's location
	 * @return
	 * @author Xing Jiankuan
	 */
	private TupleTypeInfo[] getTupleTypeInfo(Location target) {
		RemTupleTypeInfoOp op = 
			new RemTupleTypeInfoOp(dts.getName(), target.getID());
		cMgr.postOp(op, false);
		op.getResult().lock.hold();
		//wait the RemTupleTypeInfoExchange Protocol release this lock
		return (TupleTypeInfo[])op.getResult().get();
	}
	
	
	private void gatherAllRemoteInfo()
	{
		HashMap<TupleType,LinkedList<TupleTypeInfo>> infosMap 
		= new HashMap<TupleType,LinkedList<TupleTypeInfo>>();
		            
		Location[] locations  = dts.getAllEngagedLocations();
	    for(int i = 0 ; i< locations.length ;i++)
	    {
	    	if(locations[i].isHere()) 
	    		continue;
	    	TupleTypeInfo[] infos = getTupleTypeInfo(locations[i]);
	    	if(infos != null)
	    	{
	    	for(int j = 0; j < infos.length ; j++)
	    	{
	    		TupleTypeInfo in = infos[j];
	    		if(in == null)
	    			continue;
	    		if(!infosMap.containsKey(in.getType()))
	    			infosMap.put(in.getType(), new LinkedList<TupleTypeInfo>());
	    		infosMap.get(in.getType()).add(in);
	    	}
	    	}else
	    	System.out.println(locations[i]+" has no info");
	    	
	    }
		
	    for(TupleType type : infosMap.keySet())
	    {
	    	RemoteTypeInfo o = RemoteTypeInfo.getTypeRecord(TSName, type);
	    	o.accuRemoteInfos(infosMap.get(type));
	    }
//	    printTest(RemoteTypeInfo.getTypeRecordList(TSName));
	}
	
	private void printTest(RemoteTypeInfo[] map)
	{
		
		
			StringBuffer sb = new StringBuffer();
			for(RemoteTypeInfo i : map)
			{
				sb.append(i +"\n");
			}
			System.out.println(sb);
	}
}
